package com.codejiwei.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/5/25 14:40
 */
public class Flink_SQL_Connectors_DataGen {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(5);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.executeSql(
                "CREATE TABLE Orders (\n" +
                    "    test TIMESTAMP_LTZ,\n" +
                    "    test_1 STRING,\n" +
                    "    order_number BIGINT,\n" +
                    "    price        DECIMAL(32,2),\n" +
                    "    buyer        ROW<first_name STRING, last_name STRING>,\n" +
                    "    order_time   ARRAY<INT>\n" +
                    ") WITH (\n" +
                    "  'connector' = 'datagen',\n" +
//                    "  'rows-per-second' = '5',\n" +
                    "  'number-of-rows' = '10',\n" +
                    "  'fields.order_number.kind' = 'sequence',\n" +
                    "  'fields.order_number.start' = '10001',\n" +
                    "  'fields.order_number.end' = '10000000',\n" +

                    "  'fields.price.kind' = 'random',\n" +
                    "  'fields.price.min' = '100.0',\n" +
                    "  'fields.price.max' = '999.9',\n" +

                    "  'fields.order_time.element.min' = '1',\n" +
                    "  'fields.order_time.element.max' = '10',\n" +

                    "  'fields.test_1.length' = '6',\n" +

                    "  'fields.buyer.first_name.length' = '5',\n" +
                    "  'fields.buyer.last_name.length' = '5'\n" +
                    ")");

//        tableEnv.executeSql("CREATE TABLE print_table WITH (" +
//                "'connector' = 'print',\n" +
//                "'print-identifier' = '==='\n" +
//                ")\n" +
//                "LIKE Orders (EXCLUDING ALL)");

        tableEnv.executeSql("create table print_table (" +
                    "str_1 STRING, \n" +
                    "str_2 STRING) \n" +
                    "with ('connector' = 'print')"
                );


//        tableEnv.executeSql("insert into print_table select  from Orders");
        tableEnv.executeSql("insert into print_table select test_1, SUBSTRING(test_1 FROM 2) from Orders");


    }
}
